package com.bop.appender;

import org.apache.logging.log4j.core.Filter;
import org.apache.logging.log4j.core.Layout;
import org.apache.logging.log4j.core.LogEvent;
import org.apache.logging.log4j.core.appender.AbstractAppender;
import org.apache.logging.log4j.core.config.Property;
import org.apache.logging.log4j.core.config.plugins.Plugin;
import org.apache.logging.log4j.core.config.plugins.PluginAttribute;
import org.apache.logging.log4j.core.config.plugins.PluginElement;
import org.apache.logging.log4j.core.config.plugins.PluginFactory;

import java.io.Serializable;
import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Author: LiFuqiang
 * @Date: 2022/1/21 15:55
 * @Version 1.0
 * @description
 */
@Plugin(name = "HdfsAppender", category = "Core", elementType = "appender", printObject = true)
public class HdfsAppender extends AbstractAppender {
    private final ScheduledExecutorService timedRollerPool; // 定时调度线程池
    private final Layout<? extends Serializable> layout;
    private HdfsAppenderLogCache hdfsAppenderLogCache;
    private long rollInterval;
    private long maxIdleTime;
    private HdfsWriter hdfsWriter;

    protected HdfsAppender(String name, Filter filter, Layout<? extends Serializable> layout, boolean ignoreExceptions, Property[] properties) {
        super(name, filter, layout, ignoreExceptions, properties);
        timedRollerPool = new ScheduledThreadPoolExecutor(1);
        //timedRollerPool = new ScheduledThreadPoolExecutor(1,new Daemon.DaemonFactory());
        this.layout = layout;
    }

    @PluginFactory
    public static HdfsAppender createAppender(@PluginAttribute("name") String name,
                                              @PluginAttribute("rollInterval") String rollInterval,
                                              @PluginAttribute("maxIdleTime") String maxIdleTime,
                                              @PluginAttribute("filePath") String filePath,
                                              @PluginAttribute("fileName") String fileName,
                                              @PluginElement("Filter") final Filter filter,
                                              @PluginElement("Layout") Layout<? extends Serializable> layout,
                                              @PluginAttribute("ignoreExceptions") boolean ignoreExceptions) {
        int rollIntervalInt = rollInterval != null ? Integer.parseInt(rollInterval) : 20;
        long maxIdleTimeN = maxIdleTime != null ? Long.parseLong(maxIdleTime) : 5;
        HdfsAppender hdfsAppender = new HdfsAppender(name, filter, layout, ignoreExceptions, Property.EMPTY_ARRAY);
        hdfsAppender.setHdfsAppenderLogCache(new HdfsAppenderLogCache());
        hdfsAppender.setRollInterval(rollIntervalInt);
        hdfsAppender.setMaxIdleTime(maxIdleTimeN);

        HdfsWriter hdfsWriter = new HdfsWriter(filePath, fileName, rollIntervalInt, maxIdleTimeN);

        hdfsAppender.setHdfsWriter(hdfsWriter);
        hdfsAppender.startWriter();
        return hdfsAppender;
    }

    @Override
    public void append(LogEvent event) {
        this.hdfsAppenderLogCache.add(layout.toByteArray(event));
    }

    public void startWriter() {
        this.timedRollerPool.scheduleWithFixedDelay(new WriteDataTask(this.hdfsWriter, this.hdfsAppenderLogCache, layout), 0, 1000, TimeUnit.MILLISECONDS);
    }

    public long getRollInterval() {
        return rollInterval;
    }

    public void setRollInterval(long rollInterval) {
        this.rollInterval = rollInterval;
    }

    public long getMaxIdleTime() {
        return maxIdleTime;
    }

    public void setMaxIdleTime(long maxIdleTime) {
        this.maxIdleTime = maxIdleTime;
    }

    public HdfsAppenderLogCache getHdfsAppenderLogCache() {
        return hdfsAppenderLogCache;
    }

    public void setHdfsAppenderLogCache(HdfsAppenderLogCache hdfsAppenderLogCache) {
        this.hdfsAppenderLogCache = hdfsAppenderLogCache;
    }

    public HdfsWriter getHdfsWriter() {
        return hdfsWriter;
    }

    public void setHdfsWriter(HdfsWriter hdfsWriter) {
        this.hdfsWriter = hdfsWriter;
    }

    public static class IdleMonitorTask implements Runnable {

        @Override
        public void run() {

        }
    }

    public static class WriteDataTask implements Runnable {

        private final HdfsWriter hdfsWriter;
        private final HdfsAppenderLogCache hdfsAppenderLogCache;
        private final Layout<? extends Serializable> layout;

        public WriteDataTask(HdfsWriter hdfsWriter, HdfsAppenderLogCache hdfsAppenderLogCache, Layout<? extends Serializable> layout) {
            this.hdfsAppenderLogCache = hdfsAppenderLogCache;
            this.hdfsWriter = hdfsWriter;
            this.layout = layout;
        }

        @Override
        public void run() {
            List<byte[]> data = hdfsAppenderLogCache.drain();
            hdfsWriter.write(data);
        }
    }
}
